package net.bwie.realtime.jtp.dws.douyin.log.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.dws.douyin.log.bean.DouYinClass;
import net.bwie.realtime.jtp.dws.douyin.log.functions.DouYinWindowFunction;
import net.bwie.realtime.jtp.dws.douyin.log.functions.DouyinBeanMapFunction;
import net.bwie.realtime.jtp.utils.DorisUtil;
import net.bwie.realtime.jtp.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import java.text.SimpleDateFormat;
import java.time.Duration;

public class DouYinMinuteWindowJob {
    public static void main(String[] args) throws Exception{
        //1.执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(3000L);
        //2.数据源
        DataStream<String> kafkaDataStream = KafkaUtil.consumerKafka(env, "dws_live_room_event_log");
//        kafkaDataStream.print();
        //3.数据转换
        DataStream<String> handleStream = handle(kafkaDataStream);
        handleStream.print();
        //4.数据输出
        DorisUtil.saveToDoris(handleStream,
                "douyin_realtime_report","dws_douyin_window_report");

        env.execute("DouYinMinuteWindowJob");
    }

    private static DataStream<String> handle(DataStream<String> stream) {
//        //1.封装实体类
//        SingleOutputStreamOperator<DouYinClass> beanStream = stream.map(new DouyinBeanMapFunction());
//        //2.设置水位线
//        SingleOutputStreamOperator<DouYinClass> timeStream = beanStream.assignTimestampsAndWatermarks(
//                WatermarkStrategy
//                        .<DouYinClass>forBoundedOutOfOrderness(Duration.ofSeconds(5))
//                        .withTimestampAssigner(new SerializableTimestampAssigner<DouYinClass>() {
//                            @Override
//                            public long extractTimestamp(DouYinClass douYinClass, long l) {
//                                return douYinClass.getTimestamp();
//                            }
//                        })
//        );
//        //3.分组
//        KeyedStream<DouYinClass, String> keyByStream = timeStream.keyBy(bean -> "");
//        //4.设置窗口大小,窗口大小1分钟
//        WindowedStream<DouYinClass, String, TimeWindow> windowStream =
//                keyByStream.window(TumblingEventTimeWindows.of(Time.milliseconds(1)));
//        //5.转换类型
//        SingleOutputStreamOperator<String> applyStream = windowStream.apply(new DouYinWindowFunction());
////        applyStream.print();
//
//        return applyStream;
        return stream.map(jsonStr -> {
            JSONObject obj = JSON.parseObject(jsonStr);

            // event_properties
            JSONObject eventProps = obj.getJSONObject("event_properties");
            String payStatus = eventProps != null ? eventProps.getString("pay_status") : "unknown";
            Integer orderQty = eventProps != null ? eventProps.getInteger("order_qty") : 0;
            Double amount = eventProps != null ? eventProps.getDouble("amount") : 0.0;

            // Doris 表字段顺序拼接成 CSV
            return String.join(",",
                    nvl(obj.getString("live_room_id")),
                    nvl(obj.getString("event_time")),
                    nvl(obj.getString("sku_id")),
                    nvl(obj.getString("anchor_id")),
                    nvl(obj.getString("platform")),
                    nvl(obj.getString("region")),
                    nvl(obj.getString("start_time")),
                    nvl(obj.getString("end_time")),
                    String.valueOf(obj.getInteger("live_duration_sec") == null ? 0 : obj.getInteger("live_duration_sec")),
                    nvl(obj.getString("product_name")),
                    String.valueOf(obj.getDouble("product_price") == null ? 0.0 : obj.getDouble("product_price")),
                    nvl(obj.getString("img_url")),
                    nvl(obj.getString("source_id")),
                    nvl(obj.getString("source_type")),
                    nvl(obj.getString("user_id")),
                    nvl(obj.getString("event_type")),
                    nvl(obj.getString("window_start")),
                    nvl(obj.getString("window_end")),
                    nvl(payStatus),
                    String.valueOf(orderQty),
                    String.valueOf(amount)
            );
        });
    }

    private static String nvl(String val) {
        return val == null ? "" : val;
    }
}
